package com.wsg.dep.queue;

import com.wsg.dep.core.event.EventFactory;
import com.wsg.dep.demo.LongEvent;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 优化后的生产者-消费者模型（基于门控机制）
 */
public class ProducerConsumerModel {

    static class LongEventFactory implements EventFactory<LongEvent> {
        private long time = System.currentTimeMillis();

        @Override
        public LongEvent newInstance() {
            LongEvent longEvent = new LongEvent();
            longEvent.setTime(time);
            return longEvent;
        }
    }

    // ====================== 配置参数 ======================
    private static final int BUFFER_SIZE = 8;    // 必须为2的幂
    private static final int PRODUCER_COUNT = 1;
    private static final int CONSUMER_COUNT = 1;

    public static void main(String[] args) throws InterruptedException {
        // 初始化队列和门控
        BaseRingQueue<LongEvent> ringBuffer = new BaseRingQueue<>(
                new LongEventFactory(), BUFFER_SIZE
        );
        ProducerConsumerBarrier barrier = new ProducerConsumerBarrier(BUFFER_SIZE);

        // 创建线程池
        ExecutorService executor = Executors.newFixedThreadPool(
                PRODUCER_COUNT + CONSUMER_COUNT
        );

        // 启动消费者
        for (int i = 0; i < CONSUMER_COUNT; i++) {
            executor.submit(new Consumer(ringBuffer, barrier, i));
        }

//        Thread.sleep(100);
        // 启动生产者
        for (int i = 0; i < PRODUCER_COUNT; i++) {
            executor.submit(new Producer(ringBuffer, barrier, i));
        }


    }

    // ====================== 生产者实现 ======================
    static class Producer implements Runnable {
        private final BaseRingQueue<LongEvent> ringBuffer;
        private final ProducerConsumerBarrier barrier;
        private final int id;

        public Producer(BaseRingQueue<LongEvent> ringBuffer,
                        ProducerConsumerBarrier barrier, int id) {
            this.ringBuffer = ringBuffer;
            this.barrier = barrier;
            this.id = id;
        }

        @Override
        public void run() {
//            while (!Thread.currentThread().isInterrupted()) {
            for (int i = 0; i < Test.total; i++) {
                // 1. 申请写入位置
                long sequence = barrier.claimProducerSequence();

//                try {
//                    Thread.sleep(1000);
//                } catch (InterruptedException e) {
//                    e.printStackTrace();
//                }

                // 2. 写入数据
                LongEvent event = ringBuffer.get(sequence);
                event.set(i);
                barrier.commitProduceSequence(sequence);
//                System.out.printf("[Producer-%d] Write [%d]%n", id, sequence);
            }
        }
    }

    // ====================== 消费者实现 ======================
    static class Consumer implements Runnable {
        private final BaseRingQueue<LongEvent> ringBuffer;
        private final ProducerConsumerBarrier barrier;
        private final int id;
        private long nextSequence = 0L; // 消费者私有序列号

        public Consumer(BaseRingQueue<LongEvent> ringBuffer,
                        ProducerConsumerBarrier barrier, int id) {
            this.ringBuffer = ringBuffer;
            this.barrier = barrier;
            this.id = id;
        }

        @Override
        public void run() {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    // 1. 等待可用数据范围
                    long availableSequence = barrier.waitForConsumer(nextSequence);

                    // 2. 批量处理可用数据
                    while (nextSequence <= availableSequence) {
                        LongEvent event = ringBuffer.get(nextSequence);
                        process(event);
//                        System.out.printf("[Consumer-%d] Process [%d], EVENT_numb[%d]%n", id, nextSequence, event.get());
                        // 3. 提交消费进度（//实时提交消费进度）
                        barrier.commitConsumerSequence(nextSequence);
                        nextSequence++;
                    }

                    // 3. 提交消费进度
//                    barrier.commitConsumerSequence(nextSequence - 1);

                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }

        private void process(LongEvent event) throws InterruptedException {
            Test.doSomeThing(event);
        }
    }
}